package com.inventive.jmscache.server.hazelcast;

import com.hazelcast.nio.DataSerializable;

import java.io.*;
import java.util.Date;

/**
 * Created by IntelliJ IDEA.
 */
public class DataSerializationHelper {

    public static void writeSafeUTF(DataOutput dataOutput, String utf) throws IOException {
        if (null == utf) {
            dataOutput.writeBoolean(true);
        } else {
            dataOutput.writeBoolean(false);
            dataOutput.writeUTF(utf);
        }
    }

    public static String readSafeUTF(DataInput dataInput) throws IOException {
        if (!dataInput.readBoolean()) {
            return dataInput.readUTF();
        }
        return null;
    }

    public static void writeSafeDate(DataOutput dataOutput, Date date) throws IOException {
        if (null == date) {
            dataOutput.writeBoolean(true);
        } else {
            dataOutput.writeBoolean(false);
            dataOutput.writeLong(date.getTime());
        }
    }

    public static Date readSafeDate(DataInput dataInput) throws IOException {
        if (!dataInput.readBoolean()) {
            return new Date(dataInput.readLong());
        }
        return null;
    }

    public static void writeSafeSerializable(DataOutput dataOutput, Serializable serializable) throws IOException {
        if (null == serializable) {
            dataOutput.writeBoolean(true);
        } else {
            dataOutput.writeBoolean(false);
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(outputStream);
            oos.writeObject(serializable);
            oos.close();
            outputStream.close();
            byte[] bytes = outputStream.toByteArray();
            dataOutput.write(bytes.length);
            dataOutput.write(bytes);
        }
    }

    public static Serializable readSafeSerializable(DataInput dataInput) throws IOException {
        if (!dataInput.readBoolean()) {
            int size = dataInput.readInt();
            byte[] bytes = new byte[size];
            ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
            try {
                Serializable serializable = (Serializable)ois.readObject();
                ois.close();
                return serializable;
            } catch (ClassNotFoundException e) {
                throw new RuntimeException(e);
            }
        }
        return null;
    }

    public static void writeSafeDataSerializable(DataOutput dataOutput, DataSerializable serializable) throws IOException {
        if (null == serializable) {
            dataOutput.writeBoolean(true);
        } else {
            dataOutput.writeBoolean(false);
            dataOutput.writeUTF(serializable.getClass().getName());
            serializable.writeData(dataOutput);
        }
    }

    public static DataSerializable readSafeDataSerializable(DataInput dataInput) throws IOException {
        if (!dataInput.readBoolean()) {
            String classname = dataInput.readUTF();
            try {
                Class klass = Class.forName(classname);
                DataSerializable serializable = (DataSerializable)klass.newInstance();
                serializable.readData(dataInput);
                return serializable;
            } catch (ClassNotFoundException ex) {
                throw new RuntimeException(ex);
            } catch (IllegalAccessException ex) {
                throw new RuntimeException(ex);
            } catch (InstantiationException ex) {
                throw new RuntimeException(ex);
            }
        }
        return null;
    }

    public static void writeSafeObject(DataOutput dataOutput, Object object) throws IOException {
        if (null == object) {
            dataOutput.writeBoolean(true);
        } else {
            dataOutput.writeBoolean(false);
            if (object instanceof DataSerializable) {
                dataOutput.writeInt(SerializationType.DATA_SERIALIZABLE.ordinal());
                writeSafeDataSerializable(dataOutput, (DataSerializable)object);
            } else {
                dataOutput.writeInt(SerializationType.SERIALIZABLE.ordinal());
                writeSafeSerializable(dataOutput, (Serializable)object);
            }
        }
    }

    public static Object readSafeObject(DataInput dataInput) throws IOException {
        if (!dataInput.readBoolean()) {
            SerializationType type = SerializationType.getValueForOrdinal(dataInput.readInt());
            switch (type) {
                case DATA_SERIALIZABLE: {
                    return readSafeDataSerializable(dataInput);
                }
                case SERIALIZABLE: {
                    return readSafeSerializable(dataInput);
                }
            }
            return dataInput.readUTF();
        }
        return null;
    }

    private enum SerializationType {
        DATA_SERIALIZABLE,
        SERIALIZABLE;

        public static SerializationType getValueForOrdinal(int ordinal) {
            return SerializationType.values()[ordinal];
        }
    }

}
